package com.shujia.flink.kafka;

import com.alibaba.fastjson2.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author shujia
 */
public class Demo1KafkaSource {
    public static void main(String[] args) throws Exception {
        //创建flink环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建kafka source  -- 无界流
        KafkaSource<String> kafkaSource = KafkaSource
                .<String>builder()
                //指定kafka 集群列表
                .setBootstrapServers("master:9092,node1:9092,node2:9092")
                //指定消费的topic
                .setTopics("student")
                //执行消费者组
                .setGroupId("Demo1KafkaSource")
                //指定读取数据的位置,earliest: 从最早开始消费
                .setStartingOffsets(OffsetsInitializer.earliest())
                //执行反序列化类
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        //使用kafka source
        DataStream<String> lines = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka source");

        //解析数据
        DataStream<Student> students = lines.map(line -> JSON.parseObject(line, Student.class), TypeInformation.of(Student.class));

        //转换成kv
        DataStream<Tuple2<String, Integer>> kvs = students.map(stu -> Tuple2.of(stu.getClazz(), 1), Types.TUPLE(Types.STRING, Types.INT));

        //统计班级的认识
        DataStream<Tuple2<String, Integer>> counts = kvs.keyBy(kv -> kv.f0).sum(1);

        counts.print();

        env.execute();

    }

    @Data
    @AllArgsConstructor
    static class Student {
        private String id;
        private String name;
        private Integer age;
        private String gender;
        private String clazz;
    }
}
